package com.xl.flinkdemo.聚合算子;

import com.xl.flinkdemo.entity.SensorReading;
import com.xl.flinkdemo.wc.WordCount;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 基于流处理操作的API
 **/
public class RollingAggregationCount {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 从文件中读取数据

        env.setParallelism(1);

        String inputPath = "D:\\myWork\\flinkdemo\\src\\main\\resources\\sensors";
        DataStream<String> inputdataStream = env.readTextFile(inputPath);


        // 基于数据流进行转换计算
        DataStream<SensorReading> dataStream =
                inputdataStream.map(new MapFunction<String, SensorReading>() {
                    @Override
                    public SensorReading map(String s) throws Exception {
                        String[] fields = s.split(",");
                        return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                    }
                });

        //分组,按照第一个属性分组
        // 报错：KeyedStream<SensorReading, Tuple> keybyStream = dataStream.keyBy(0);
        //keyBy有多个重载方法
        KeyedStream<SensorReading, Tuple> keybyStream = dataStream.keyBy("id");
        //等同于 dataStream.keyBy(data -> data.getId());

        // 滚动聚合max(),min().maxby().minby() s使用滚动聚合的API，必须要使用keyby

        //取当前最大的温度值
        SingleOutputStreamOperator<SensorReading> resultStream = keybyStream.maxBy("sensors");



        resultStream.print();

        env.execute();
    }
}
